package cn.tedu.charging.order.rabbitmq;

import cn.tedu.charging.common.constant.AmqpComponentConst;
import cn.tedu.charging.common.pojo.message.DelayCheckMessage;
import cn.tedu.charging.common.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.common.protocol.WebSocketMessage;
import cn.tedu.charging.order.feign.DeviceClient;
import cn.tedu.charging.order.mapper.BillFailMapper;
import cn.tedu.charging.order.mapper.BillSuccessMapper;
import cn.tedu.charging.order.websocket.WebSocketServer;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Date;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class AmqpConsumer {
    @Resource
    private BillSuccessMapper billSuccessMapper;
    @Resource
    private BillFailMapper billFailMapper;
    @Resource
    private RedisTemplate<String,String> redisTemplate;
    @Resource
    private DeviceClient deviceClient;
    @Resource
    private WebSocketServer webSocketServer;
    // 对接死信队列 消费延迟消息
    @RabbitListener(queues = AmqpComponentConst.DELAY_BILL_DL_QUEUE)
    public void consumeDelayBill(DelayCheckMessage msg, Message message, Channel channel) {
        boolean ack = false;
        String lockKey = "charging:bill:lock:" + msg.getOrderNo();
        String value = "rabbitmq of key";
        ValueOperations<String, String> ops = redisTemplate.opsForValue();
        try {
            // 抢锁  do while 循环 直到抢到锁为止
            boolean lock; int count = 0;
            do {
                count++;
                if (count > 1) {
                    Thread.sleep(1000);
                }
                lock = Boolean.TRUE.equals(ops.setIfAbsent(lockKey, value, 5, TimeUnit.MINUTES));
            } while(!lock);
            handleCheckBill(msg);// 消费核心
            log.info("消费消息:{}", msg);
            // 手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            ack = true;
        } catch (Exception e) {
            // 处理延迟消息出现异常 手动确认重新发送消息 重新处理3次以上 拒绝处理 人工补偿
            log.error("消费消息失败:{}", msg, e);
        } finally {
            redisTemplate.delete(lockKey);
        }
        try {
            if (!ack) {
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
            }
        } catch (IOException e) {
            log.error("消息拒绝失败:{}", msg, e);
        }
    }
    private void handleCheckBill(DelayCheckMessage msg) {
        QueryWrapper queryWrapper = new QueryWrapper();
        queryWrapper.eq("bill_id", msg.getOrderNo());
        Long success = billSuccessMapper.selectCount(queryWrapper);
        Long fail = billFailMapper.selectCount(queryWrapper);
        if (success > 0 || fail > 0) {
            return;
        } else {
            ChargingBillFailPO po = new ChargingBillFailPO();
            // 订单编号 用户id 时间
            po.setBillId(msg.getOrderNo());
            po.setUserId(msg.getUserId());
            po.setCreateTime(new Date());
            po.setFailDesc("设备无响应");
            billFailMapper.insert(po);
            // 设计一个修改枪状态的device服务接口
            deviceClient.changeGunStatus(msg.getGunId(), 4);
            // 实时消息推送 告知设备故障 或者其他原因 订单无法继续充电 请换枪 赠送优惠券
            WebSocketMessage<String> wsm = new WebSocketMessage<>();
            // state=1表示订单有问题 2用户余额不足 3正常同步充电进度
            wsm.setState(1);
            wsm.setMessage("订单失败 请换枪");
            wsm.setData("B扫码下单失败，请换枪，非常抱歉，赠送您一张优惠券:http：//AN0df32fj3");
            webSocketServer.sendMessage(msg.getUserId(),wsm);
        }
    }
}
